/*
 * reader.c - EBCD PCH reader
 * Clunix, KyungWoon Cho, cezanne@clunix.com, 2006.2
 *
 * Copyright (c) 2001-2008 by Clunix Inc. All Rights Reserved.
 */

#include "config.h"

#include "ebcd.h"

extern ecm_mutex_t	accepted_mutex;
extern struct list_head	accepted_pchs;

static ecm_uint32	n_receivers;

static ecm_mutex_t	receiver_mutex;
static ecm_cond_t	receiver_empty;

static ECMBOOL		stopping_receivers;

static void
do_receive(ecm_ptr_t arg)
{
	ebcd_pch_t	*dpch = (ebcd_pch_t *)arg;
	ebc_pch_t	*pchp = dpch->pchp;
	ebc_pktb_t	*pktb;
	ecm_uint32	nrecv = 0;
	ebc_err_t	err = EBC_OK;

	while (!stopping_receivers &&
	       (err = pch_read_pktb(pchp, &pktb)) == EBC_OK) {
		ECMASSERT(pktb->pkthdr.mhdr.dstaddr.ndid != EBC_NDID_SELF &&
			  pktb->pkthdr.mhdr.srcaddr.ndid != EBC_NDID_SELF);

		nrecv++;
		ebcd_add_lazy_pktb(dpch, pktb);
	}

	ecm_mutex_lock(receiver_mutex);
	ECMASSERT(n_receivers > 0);
	n_receivers--;
	if (n_receivers == 0)
		ecm_cond_signal(receiver_empty);
	ecm_mutex_lock(pchp->mutex);
	pchp->receiver = ECM_FALSE;
	ecm_mutex_unlock(pchp->mutex);
	ecm_mutex_unlock(receiver_mutex);

	if (nrecv > 0)
		set_MRU_recv_dpch(dpch);

	if (!stopping_receivers && err == -ERR_EBC_DISCONN) {
		EBCDDBG(DBG_DPCH, "%s: diconnected while receiving", ebcd_dbg_pch(dpch));
		ebcd_destroy_pch(dpch);
	}
	ebcd_pch_put(dpch);
}

void
ebcd_read_pchs(void)
{
	ebcd_pch_t	*dpch;
	ebc_pch_t	*pchp;
	struct list_head	*lp;

	ecm_mutex_lock(accepted_mutex);

	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);
		pchp = dpch->pchp;

		if (!PCH_ISSET_EVENT(pchp, ebcd_selection, ECM_READ_EVENT))
			continue;
		if (!ebcd_pch_get(dpch))
			continue;

		ecm_mutex_lock(receiver_mutex);
		ecm_mutex_lock(pchp->mutex);
		if (pchp->receiver) {
			ecm_mutex_unlock(pchp->mutex);
			ecm_mutex_unlock(receiver_mutex);
			ebcd_pch_put(dpch);
			continue;
		}

		pchp->receiver = ECM_TRUE;
		if (!ecm_start_worker(ebcd_core_workers, do_receive, dpch)) {
			pchp->receiver = ECM_FALSE;
			ecm_mutex_unlock(pchp->mutex);
			ecm_mutex_unlock(receiver_mutex);
			ebcd_pch_put(dpch);
		}
		else {
			ecm_mutex_unlock(pchp->mutex);
			n_receivers++;
			ecm_mutex_unlock(receiver_mutex);
		}
	}

	ecm_mutex_unlock(accepted_mutex);
}

void
ebcd_wait_reader(void)
{
	ecm_uint32	timeout = ebcd_alivecheck_timeout;

	ecm_mutex_lock(receiver_mutex);
	while (n_receivers > 0) {
		if (!ecm_cond_wait(receiver_empty, receiver_mutex, &timeout)) {
			stopping_receivers = ECM_TRUE;
			break;
		}
	}
	while (n_receivers > 0)
		ecm_cond_wait(receiver_empty, receiver_mutex, NULL);
	ecm_mutex_unlock(receiver_mutex);
	stopping_receivers = ECM_FALSE;
}

void
ebcd_start_reader(void)
{
	receiver_mutex = ecm_mutex_create(ECM_FALSE);
	receiver_empty = ecm_cond_create();
}

void
ebcd_stop_reader(void)
{
	ecm_mutex_free(receiver_mutex);
	ecm_cond_free(receiver_empty);
}
